package org.xq.softcup.mq.remoting.net.impl.netty.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.xq.softcup.mq.remoting.invoker.XQRpcInvokerFactory;
import org.xq.softcup.mq.remoting.net.common.ConnectClient;
import org.xq.softcup.mq.remoting.net.impl.netty.codec.NettyDecoder;
import org.xq.softcup.mq.remoting.net.impl.netty.codec.NettyEncoder;
import org.xq.softcup.mq.remoting.net.params.XQRpcRequest;
import org.xq.softcup.mq.remoting.net.params.XQRpcResponse;
import org.xq.softcup.mq.serialize.Serializer;
import org.xq.softcup.mq.util.IpUtil;

@Slf4j
public class NettyConnectClient extends ConnectClient {
    private EventLoopGroup group;
    private Channel channel;

    public void init(String address, final Serializer serializer, final XQRpcInvokerFactory xqRpcInvokerFactory)
            throws Exception {
        Object[] array = IpUtil.parseIpPort(address);
        String host = (String) array[0];
        int port = ((Integer) array[1]).intValue();

        this.group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(this.group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer() {
                    @Override
                    protected void initChannel(Channel channel) throws Exception {
                        channel.pipeline()
                                .addLast(new ChannelHandler[]{new NettyEncoder(XQRpcRequest.class, serializer)})
                                .addLast(new ChannelHandler[]{new NettyDecoder(XQRpcResponse.class, serializer)})
                                .addLast(new ChannelHandler[]{new NettyClientHandler(xqRpcInvokerFactory)});
                    }
                })
                .option(ChannelOption.TCP_NODELAY, Boolean.valueOf(true))
                .option(ChannelOption.SO_KEEPALIVE, Boolean.valueOf(true))
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, Integer.valueOf(10000));

        this.channel = bootstrap.connect(host, port).sync().channel();
        if (!isValidate()) {
            close();
            return;
        }
        log.debug("Netty客户端代理, 连接Server成功！ host:{}, port:{}", host, Integer.valueOf(port));
    }

    public boolean isValidate() {
        if (this.channel != null) {
            return this.channel.isActive();
        }
        return false;
    }

    public void close() {
        if ((this.channel != null) && (this.channel.isActive())) {
            this.channel.close();
        }
        if ((this.group != null) && (!this.group.isShutdown())) {
            this.group.shutdownGracefully();
        }
        log.debug("netty 客户端关闭.");
    }

    public void send(XQRpcRequest xqRpcRequest)
            throws Exception {
        this.channel.writeAndFlush(xqRpcRequest).sync();
    }
}
